package com.zhang.limit;

import com.google.common.util.concurrent.Monitor;
import com.google.common.util.concurrent.RateLimiter;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.stream.IntStream;

/**
 * 漏洞算法
 */
public class LeakBucketTest {
    public static void main(String[] args) {
        final LeakBucket leakBucket = new LeakBucket();
        final AtomicInteger DATA_CREATOR = new AtomicInteger(0);
        // 生产线程 10个线程 每秒提交 50个数据  1/0.2s*10=50个
        IntStream.range(0, 50).forEach(i -> {
            new Thread(() -> {
                for (; ; ) {
                    int data = DATA_CREATOR.incrementAndGet();
                    try {
                        leakBucket.submit(data);
                        TimeUnit.MILLISECONDS.sleep(200);
                    } catch (Exception e) {
                        //对submit时，如果桶满了可能会抛出异常
                        if (e instanceof IllegalStateException) {
                            System.out.println(e.getMessage());
                            //当满了后，生产线程就休眠1分钟
                            try {
                                TimeUnit.SECONDS.sleep(60);
                            } catch (InterruptedException e1) {
                                e1.printStackTrace();
                            }
                        }
                    }
                }
            }).start();
        });

        // 消费线程  采用RateLimiter每秒处理10个  综合的比率是5:1
        IntStream.range(0, 10).forEach(i -> {
            new Thread(
                    () -> {
                        for (; ; ) {
                            leakBucket.takeThenConsumer(x -> {
                                System.out.println(Thread.currentThread() + " 消费:" + x);
                            });
                        }
                    }
            ).start();
        });
    }

    /**
     * 漏洞
     */
    static class LeakBucket {
        // 定义桶的大小
        private final ConcurrentLinkedQueue<Integer> container = new ConcurrentLinkedQueue<>();

        private final static int BUCKET_LIMIT = 50;

        // 消费者 不论多少个线程，每秒最大的处理能力是1秒中执行10次
        private final RateLimiter consumerRate = RateLimiter.create(10d);

        // 往桶里面放数据时，确认没有超过桶的最大的容量
        private final Monitor offerMonitor = new Monitor();

        // 从桶里消费数据时，桶里必须存在数据
        private final Monitor consumerMonitor = new Monitor();


        /**
         * 往桶里面写数据
         *
         * @param data 数据
         */
        public void submit(Integer data) {
            if (offerMonitor.enterIf(offerMonitor.newGuard(() -> container.size() < BUCKET_LIMIT))) {
                try {
                    container.offer(data);
                    System.out.println(Thread.currentThread() + " 提交:" + data + " 当前容器的大小 :[" + container.size() + "]");
                } finally {
                    offerMonitor.leave();
                }
            } else {
                // 这里时候采用降级策略了。消费速度跟不上产生速度时，而且桶满了，抛出异常
                // 或者存入MQ DB等后续处理
                throw new IllegalStateException(Thread.currentThread().getName() + "当前桶已经满了，请稍后再试");
            }
        }


        /**
         * 从桶里面消费数据
         *
         * @param consumer 消费者
         */
        public void takeThenConsumer(Consumer<Integer> consumer) {
            if (consumerMonitor.enterIf(consumerMonitor.newGuard(() -> !container.isEmpty()))) {
                try {
                    // 不打印时 写 consumerRate.acquire();
                    System.out.println(Thread.currentThread() + "  获取数据等待的时间 " + consumerRate.acquire());
                    Integer data = container.poll();
                    // container.peek() 只是去取出来不会删掉
                    consumer.accept(data);
                } finally {
                    consumerMonitor.leave();
                }
            } else {
                // 当木桶的消费完后，可以消费那些降级存入MQ或者DB里面的数据
                System.out.println("容器中数据已经消费完成，可与从MQ或者DB消费数据...");
                try {
                    TimeUnit.SECONDS.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
